package org.example.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class WordCountStreamUnbound {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
        //本地IDE调试，用显示WEB UI界面
        //StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //全局禁用算子链
        DataStreamSource<String> stringDataStreamSource = env.socketTextStream("192.168.99.1", 9999);
        stringDataStreamSource.
                flatMap((FlatMapFunction<String, String>) (value, out) -> {
                    String[] split = value.split(" ");
                    for (String s : split) {
                        out.collect(s);
                    }
                })
                .returns(Types.STRING)
                .map(value -> Tuple2.of(value, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT)) //解析Lambda类型擦除的问题。
                .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0).sum(1).print();
        env.execute("WordCountStreamUnbound");
    }
}
